package com.bff.gaia.streaming.api.xjoin.core.side;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.streaming.api.functions.async.ResultFuture;
import com.bff.gaia.streaming.api.functions.async.RichAsyncFunction;
import com.bff.gaia.streaming.api.xjoin.core.enums.ECacheType;
import com.bff.gaia.streaming.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.LRUSideCache;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.AbsSideCache;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.types.Row;

import java.util.Collections;

/**
 * Description: gaia-parent
 * Created by tiger on 2019/7/19 15:01.
 */
public abstract class AsyncReqRow extends RichAsyncFunction<Row,Row> implements ISideReqRow {

	public static final long serialVersionUID = 2098635244857937717L;

	protected SideInfo sideInfo;

	public AsyncReqRow(SideInfo sideInfo){
		this.sideInfo = sideInfo;
	}

	private void initCache(){
		SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
		if(sideTableInfo.getCacheType() == null || ECacheType.NONE.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
			return;
		}

		AbsSideCache sideCache;
		if(ECacheType.LRU.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
			sideCache = new LRUSideCache(sideTableInfo);
			sideInfo.setSideCache(sideCache);
		}else{
			throw new RuntimeException("not support side cache with type:" + sideTableInfo.getCacheType());
		}

		sideCache.initCache();
	}

	protected CacheObj getFromCache(String key){
		return sideInfo.getSideCache().getFromCache(key);
	}

	protected void putCache(String key, CacheObj value){
		sideInfo.getSideCache().putCache(key, value);
	}

	protected boolean openCache(){
		return sideInfo.getSideCache() != null;
	}

	protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
		if(sideInfo.getJoinType() == EJoinType.LEFT){
			//Reserved left table data
			Row row = fillData(input, null);
			resultFuture.complete(Collections.singleton(row));
		}else{
			resultFuture.complete(null);
		}
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		initCache();
	}

	@Override
	public void close() throws Exception {
		super.close();
	}


}